package com.game.gateway;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

//@Service //测试的时候打开注释
public class KafkaBusTest {
    @Autowired
    private KafkaTemplate<String, byte[]> kafkaTemplate;

    @PostConstruct
    public void init() throws ExecutionException, InterruptedException {
        System.out.println("初始化。。。。。。。");
        String str = "你好，kafka";
        ProducerRecord<String, byte[]> record = new ProducerRecord<String, byte[]>("springCloudBus", "hello", str.getBytes());
        Future future = kafkaTemplate.send(record);
        future.get();
    }
    @KafkaListener(topics = {"springCloudBus"},groupId = "game-kafka-cluster")
    public void receiver(ConsumerRecord<String, byte[]> record) {
        byte[] body = record.value();
        String value = new String(body);
        System.out.println("收到kafka的消息：" + value);
    }
}
